package com.MR;

import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;

import java.io.File;
import java.io.IOException;

/**
 * 小文件解决方案之SequenceFile容器
 */
public class SmallFileSequenceFile {

    public static void main(String[] args) throws IOException {
        //生成SequenceFile文件
        write("D:\\smallFile","/SeqFile");
        //读取SequenceFile文件
        read("/SeqFile");
    }
    /**
     * 生成SequenceFile文件
     * @param inputDir 输入目录
     * @param outputFile 输出文件HDFS文件
     * @throws IOException
     */
    private static void write(String inputDir,String outputFile) throws IOException {
        //1.创建一个配置对象指定HDFS对应的地址
        Configuration configuration=new Configuration();
        configuration.set("fs.defaultFS","hdfs://bigdata01:9000");

        //2.根据配置信息获取操作HDFS的对象来删除HDFS上的输出文件--实现重复执行
        FileSystem fileSystem=FileSystem.get(configuration);
        fileSystem.delete(new Path(outputFile),true);

        //3.创建SequenceFile.Writer.Option的数组对象
        //  封装三个信息--输出路径、key类型、value类型
        SequenceFile.Writer.Option[] opts = new SequenceFile.Writer.Option[]{
            SequenceFile.Writer.file(new Path(outputFile)), //指定输出路径
            SequenceFile.Writer.keyClass(Text.class),//指定Key类型
            SequenceFile.Writer.valueClass(Text.class)//指定value类型
        };
        //4. 创建一个Writer实例
        SequenceFile.Writer writer=SequenceFile.createWriter(configuration,opts);

        //5. 指定需要压缩的文件的目录，并将目录中的文件名和内容作为键值对写入SequenceFile
        File inputDirPath=new File(inputDir);
        //判断输入的路径地址是否是一个目录
        if (inputDirPath.isDirectory()){
            //获取目录中的所有文件
            File [] files=inputDirPath.listFiles();
            //迭代文件
            for (File file :files) {
                //获取文件中的所有内容
                String content= FileUtils.readFileToString(file,"UTF-8");
                //获取文件名
                String fileName=file.getName();
                //将文件名、文件内容作为键值对，向SequenceFile中写入该键值对数据
                Text key=new Text(fileName);
                Text value=new Text(content);
                writer.append(key,value);
            }
        }
        //关闭写入流
        writer.close();
    }

    /**
     * 读取SequenceFile文件
     * @param inputFile SequenceFile文件路径
     * @throws IOException
     */
    private static void read(String inputFile) throws IOException {
        //1.创建一个配置对象指定HDFS对应的地址
        Configuration configuration=new Configuration();
        configuration.set("fs.defaultFS","hdfs://bigdata01:9000");
        //2. 创建SequenceFile.Reader的阅读器对象
        SequenceFile.Reader reader=new SequenceFile.Reader(configuration,SequenceFile.Reader.file(new Path(inputFile)));
        //3. 创建两个Text对象来保存读取的键值对信息
        Text key=new Text();
        Text value=new Text();
        //4. 循环读取数据，数据保存在key、value中
        while(reader.next(key,value)){
            //输出文件名和文件内容
            System.out.println("文件名:"+key.toString()+"----文件内容："+value.toString());
        }
    }

}
